package com.bw.dwd;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.DateFormatUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * @author y3068
 * @Classname Test1
 * @Description TODO
 * @Date 2025/6/20 21:07
 * @Version V1.0
 */

public class Test1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> ds = env.addSource(MyKafkaUtil.getFlinkKafkaConsumerAll("topic_log", "base_log_consumer"));
//        ds.print();

        OutputTag<String> dirtyStreamTag = new OutputTag<>("dirtyStream");

        SingleOutputStreamOperator<String> ds1 = ds.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                if (JSON.isValid(s)) {
                    collector.collect(s);
                } else {
                    context.output(dirtyStreamTag, s);
                }
            }
        });

        DataStream<String> dirtyStream = ds1.getSideOutput(dirtyStreamTag);

        SingleOutputStreamOperator<JSONObject> json_data = ds1.map(JSON::parseObject);

        SingleOutputStreamOperator<JSONObject> ds2 = json_data.keyBy(t -> t.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        getRuntimeContext().getState(new ValueStateDescriptor<String>("firstView", String.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        String isNew = jsonObject.getJSONObject("common").getString("is_new");
                        String value = valueState.value();
                        Long ts = jsonObject.getLong("ts");

                        String date = DateFormatUtil.toDate(ts);
                        if (isNew.equals("1")) {
                            if (valueState == null) {
                                valueState.update(date);
                            } else {
                                if (!valueState.equals(date)) {
                                    jsonObject.put(isNew, "0");
                                }
                            }
                        } else {
                            if (valueState == null) {
                                String year = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                valueState.update(year);
                            }
                        }
                        collector.collect(jsonObject);
                    }
                });


        OutputTag<String> actionTag = new OutputTag<>("action");
        OutputTag<String> errTag = new OutputTag<>("err");
        OutputTag<String> displaysTag = new OutputTag<>("displays");
        OutputTag<String> startTag = new OutputTag<>("start");

        SingleOutputStreamOperator<String> ds3 = ds2.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {

                JSONObject err1 = jsonObject.getJSONObject("err");
                if (err1 != null) {
                    context.output(errTag, jsonObject.toJSONString());
                }
                jsonObject.remove("err");

                JSONObject start1 = jsonObject.getJSONObject("start");
                if (start1 != null) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    JSONObject page = jsonObject.getJSONObject("page");
                    JSONObject common = jsonObject.getJSONObject("common");
                    Long ts = jsonObject.getLong("ts");

                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {

                            JSONObject display = displays.getJSONObject(i);
                            JSONObject object = new JSONObject();
                            object.put("display", display);
                            object.put("common", common);
                            object.put("page", page);
                            object.put("ts", ts);

                            context.output(displaysTag, display.toJSONString());
                        }
                    }

                    JSONArray actions = jsonObject.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject object = new JSONObject();
                            object.put("action", action);
                            object.put("common", common);
                            object.put("page", page);
                            object.put("ts", ts);
                            context.output(actionTag, action.toJSONString());
                        }
                    }

                    jsonObject.remove("displays");
                    jsonObject.remove("actions");

                    collector.collect(jsonObject.toJSONString());
                }
            }
        });

        DataStream<String> startDS = ds3.getSideOutput(startTag);
        DataStream<String> displayDS = ds3.getSideOutput(displaysTag);
        DataStream<String> actionDS = ds3.getSideOutput(actionTag);
        DataStream<String> errorDS = ds3.getSideOutput(errTag);

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        ds3.print("page>>>>>>>>>>>>>>>>>>>");
        ds3.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));

        env.execute();
    }
}
